package com.jing.cloud.client;

import com.google.common.collect.Collections2;
import com.jing.cloud.client.config.ClientConf;
import com.jing.cloud.client.factory.ClientFactory;
import com.jing.cloud.client.factory.ClientPool;
import com.jing.cloud.client.zookeeper.Node;
import com.jing.cloud.client.zookeeper.ServiceTree;
import com.jing.cloud.client.zookeeper.ZooKeeperListener;
import com.jing.cloud.service.MicroService;
import com.jing.cloud.service.Req;
import com.jing.cloud.service.Rsp;
import com.jing.cloud.service.ServiceInfo;
import com.jing.cloud.util.ErrCode;
import com.jing.cloud.util.RspUtil;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

public class ServiceUtil {

    private static final Logger log = LoggerFactory.getLogger("Thrift-Client");
    private static final ServiceTree st = ServiceTree.getInstance();
    private static final Map<String, ClientPool> clientMap = new ConcurrentHashMap<>();

    static {
        ZooKeeperListener.init();
    }

    public static Rsp call(Request request) {
        long begin = System.currentTimeMillis();
        Map<String, Node> nodeMap = st.getNodes(request.getServiceName());
        if (nodeMap.isEmpty()) {
            log.error("service [{}] unregister yet.", request.getServiceName());
            return RspUtil.error(ErrCode.SERVICE_UNREGISTER).toRsp();
        }
        Node instanceNode;
        switch (request.getRouteMode()) {
            case Request.ROUTE_MODE_RANDOM:
                Collection<Node> collection = nodeMap.values();
                if (request.getVersion() != null) {
                    collection = Collections2.filter(collection, input -> request.getVersion().equalsIgnoreCase(input.getVersion()));
                }
                if (collection.isEmpty()) {
                    log.error("service [{}#{}] not found.", request.getServiceName(), request.getVersion());
                    return RspUtil.error(ErrCode.SERVICE_NOT_FOUND).toRsp();
                }
                if (collection.size() == 1) {
                    instanceNode = collection.iterator().next();
                } else {
                    int j = new Random().nextInt(collection.size());
                    Iterator<Node> iterator = collection.iterator();
                    for (int i = 0; i < j; i++) {
                        iterator.next();
                    }
                    instanceNode = iterator.next();
                }
                break;
            case Request.ROUTE_MODE_NAMED:
                instanceNode = nodeMap.get(request.getInstanceName());
                if (null == instanceNode) {
                    log.error("service [{}@{}] not found.", request.getServiceName(), request.getInstanceName());
                    return RspUtil.error(ErrCode.SERVICE_NOT_FOUND).toRsp();
                }
                break;
            default:
                log.error("unsupported route mode");
                return RspUtil.error(ErrCode.UNSUPPORTED_ROUTE_MODE).toRsp();
        }
        ClientPool pool = getClientPool(instanceNode);
        MicroService.Client client = null;
        Rsp rsp;
        Req req = request.toReq();
        try {
            client = pool.getClient();
            if (!request.isOneway()) {
                rsp = client.callMethod(req);
            } else {
                client.callMethodOneway(req);
                rsp = RspUtil.success().toRsp();
            }
            pool.returnClient(client);
        } catch (TTransportException e) {
            log.error("callMethod req:{} error", req, e);
            rsp = RspUtil.error(ErrCode.CLIENT_ERROR, e).toRsp();
            pool.returnClient(client);
        } catch (Exception e) {
            log.error("callMethod req:{} error", req, e);
            rsp = RspUtil.error(ErrCode.CLIENT_ERROR, e).toRsp();
            if (client != null) {
                try {
                    pool.invalidateClient(client);
                } catch (Exception ex) {
                    log.error("pool.invalidateClient ", ex);
                }
            }
        }
        long end = System.currentTimeMillis();
        log.debug("callMethod use {}ms req:{} rsp:{}", (end - begin), request, rsp);
        return rsp;
    }

    private static ClientPool getClientPool(Node node) {
        synchronized (node.getKey().intern()) {
            ClientPool pool = clientMap.get(node.getKey());
            if (pool == null) {
                pool = createClientPool(node.getInfo());
                clientMap.put(node.getKey(), pool);
            }
            return pool;
        }
    }

    private static ClientPool createClientPool(ServiceInfo info) {
        log.info("create pool [{}]", info);
        String host = info.getHost();
        int port = info.getPort();
        ClientPool pool = new ClientPool();
        PooledObjectFactory<MicroService.Client> factory = new ClientFactory(host, port);
        GenericObjectPoolConfig conf = new GenericObjectPoolConfig();
        conf.setMaxTotal(ClientConf.MAX_TOTAL);
        pool.setServiceName(info.getServiceName()).initPool(factory, conf);
        return pool;
    }

}
